package com.cloudfun.msglisten.ws.haier;

import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.TypeReference;
import com.cloudfun.msglisten.common.model.dto.dingtalk.TextContent;
import com.cloudfun.msglisten.common.model.haier.HaierBaseRespDTO;
import com.cloudfun.msglisten.common.model.haier.HaierWsAckRespDTO;
import com.cloudfun.msglisten.service.DingTalkService;
import com.cloudfun.msglisten.ws.haier.config.HaierWsProperties;
import com.cloudfun.msglisten.ws.haier.handler.HaierWsCmdAckMsgHandler;
import com.cloudfun.msglisten.ws.haier.handler.HaierWsDeviceMsgHandler;
import com.google.common.util.concurrent.RateLimiter;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.Date;
import java.util.List;
import java.util.Random;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Created by Huqin on 2019/12/17
 */
@Slf4j
@Component
public class HaierWsMsgDispatcher implements DisposableBean, InitializingBean {

    private static final String ACK_MSG_FLAG = "-ack";

    // 消息处理线程池
    private ExecutorService msgHandlerExecutor;

    @Autowired
    private List<HaierWsCmdAckMsgHandler> cmdAckMsgHandlerList;

    @Autowired
    private HaierWsDeviceMsgHandler haierWsDeviceMsgHandler;

    @Autowired
    private DingTalkService dingTalkService;

    @Autowired
    private HaierWsProperties haierWsProperties;

    private volatile AtomicBoolean initlized = new AtomicBoolean(false);

    @Override
    public void afterPropertiesSet() throws Exception {
        msgHandlerExecutor = new ThreadPoolExecutor(30, 30,
                20, TimeUnit.SECONDS, new ArrayBlockingQueue<>(2000), new MyPolicy(dingTalkService,haierWsProperties));
        initlized.set(true);
    }

    public static class MyPolicy implements RejectedExecutionHandler {
        private DingTalkService dingTalkService;
        private HaierWsProperties haierWsProperties;
        //两分钟允许通知一次,防止频率过高
        private RateLimiter rateLimiter = RateLimiter.create(1/120.0D);

        public MyPolicy(DingTalkService dingTalkService,HaierWsProperties haierWsProperties){
            this.dingTalkService = dingTalkService;
            this.haierWsProperties = haierWsProperties;
        }
        @Override
        public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
            log.warn("发生消息丢弃事件");
            if (rateLimiter.tryAcquire()) {
               dingTalkService.sendMessage(new TextContent("msg-listen服务发生消息丢弃事件，请及时关注"));
            }
        }
    }

    @Override
    public void destroy() throws Exception {
        if (!msgHandlerExecutor.isShutdown()) {
            msgHandlerExecutor.shutdown();
        }
    }

    public void doDispatch(String message, HaierWsClient haierWsClient) {
        while (!initlized.get()) {

        }
        if (message.contains(ACK_MSG_FLAG)) {
            this.handleAckMsg(message, haierWsClient);
        } else {
            this.handleDeviceMsg(message);
        }
    }

    private void handleDeviceMsg(String message) {
        msgHandlerExecutor.execute(() -> {
            haierWsDeviceMsgHandler.handle(message);
        });
    }

    private void handleAckMsg(String message, HaierWsClient haierWsClient) {
        HaierWsCmdAckMsgHandler cmdAckMsgHandler = null;
        HaierBaseRespDTO<HaierWsAckRespDTO> haierBaseRespDTO = JSONObject.parseObject(message,
                new TypeReference<HaierBaseRespDTO<HaierWsAckRespDTO>>() {
                });
        for (HaierWsCmdAckMsgHandler ackMsgHandler : cmdAckMsgHandlerList) {
            String command = ackMsgHandler.command();
            if (haierBaseRespDTO.getCmd().equals(command)) {
                cmdAckMsgHandler = ackMsgHandler;
                break;
            }
        }
        if (null == cmdAckMsgHandler) {
            log.error("海尔websocket应答消息无对应handler：{}", haierBaseRespDTO.getCmd());
            return;
        }
        cmdAckMsgHandler.handleAck(haierBaseRespDTO.getData(), haierWsClient);
    }

}
